package com.mq.mqtest.threadpool;

import lombok.extern.slf4j.Slf4j;

import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @author 七弦
 * <p>
 * 线程池
 */
@Slf4j(topic = "c.ThreadPool")
public class ThreadPool {

    //阻塞队列
    private final BlockedQueue<Runnable> blockedQueue;

    //线程集合
    private final HashSet<Thread> workers = new HashSet<Thread>();

    //核心线程数
    private final int coreThreadSize;

    //超时时间
    private final long timeout;

    private final TimeUnit timeUnit;

    private RejectPolicy<Runnable> rejectPolicy; //拒绝策略

    public ThreadPool(int coreThreadSize, long timeout, TimeUnit timeUnit,
        int queueCapacity, RejectPolicy<Runnable> rejectPolicy) {
        this.coreThreadSize = coreThreadSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.blockedQueue = new BlockedQueue<>(queueCapacity);
        this.rejectPolicy = rejectPolicy;
    }

    //测试
    public static void main(String[] args) {
        RejectPolicy<Runnable> rejectPolicy = (blockedQueue, task) -> {
            //1、死等策略
//            blockedQueue.put(task);

            //2、带超时的等待
           //blockedQueue.offer(task, 1500, TimeUnit.MILLISECONDS);

            //3、放弃让调用者放弃执行任务（啥也不做就是放弃了，这里可以只打印一句话...）
            //log.error("放弃了。。。{}", task);

            //4、调用者抛出异常(和放弃执行的区别是：抛出异常可以让剩余的任务也都不执行了。
            // 比如一共4个任务，0和1任务分别被执行和放到队列；那2和3这俩任务，
            // 如果是放弃的话这俩都会放弃，
            // 如果是抛异常，2抛异常，则3根本不会被执行(主线程抛异常后，后面的任务就不会被执行了...))
            //throw new RuntimeException("task执行失败" + task);

            //5、让调用者自己执行任务
            //task.run();
        };

        ThreadPool threadPool
            = new ThreadPool(1, 1, TimeUnit.SECONDS, 1, rejectPolicy);

        //主线程向线程提交任务
        for (int i = 0; i < 4; i++) {
            //i是变化的，不能直接被lambda直接使用
            int x = i;

            threadPool.execute(() -> {
                try {
                    Thread.sleep(1000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.info("{}", x);
            });

        }
    }

    /**
     * 执行任务
     *
     * @param task
     */
    public void execute(Runnable task) {
        //如果任务数没超过核心线程数，直接交给worker对象执行
        //如果任务数超过了coreSize，就先加到任务队列暂存起来
        if (workers.size() < coreThreadSize) {
            //创建线程对象
            Worker thread = new Worker(task);
            //打印下线程和任务对象
            log.info("新增worker{}, {}", thread, task);
            //加入到线程集合中
            workers.add(thread);
            thread.start();
        } else {
//            blockedQueue.put(task);
            //1)死等
            //2)带超时的等待
            //3）让调用者放弃任务执行
            //4）让调用者抛出异常
            //5）让调用者自己执行任务

            blockedQueue.tryPut(rejectPolicy, task);
        }
    }

    class Worker extends Thread {

        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            //执行任务
            //1）当task不为空，执行任务
            //2）当task执行完毕，再从任务队列获取任务并执行
//            while (task != null || (task = blockedQueue.take()) != null) {
            while (task != null || (task = blockedQueue.poll(timeout, timeUnit)) != null) {
                try {
                    log.info("正在执行...{}", task);
                    task.run();
                } catch (Exception e) { //任务执行的时候可能会有异常，暂时捕捉一下。不作处理。
                    e.printStackTrace();
                }finally {
                    task = null;
                }
            }

            //退出循环后，移除掉线程
            synchronized (workers) {
                log.info("worker被移除{}", this);
                workers.remove(this);
            }
        }
    }
}
